package com.sbp.message.service;

import com.alibaba.fastjson.JSON;
import com.dap.dao.model.Page;
import com.dap.utils.IdGenerator;
import com.sbp.message.api.common.CommonUtils;
import com.sbp.message.api.entity.Message;
import com.sbp.message.api.entity.MessageStatus;
import com.sbp.message.api.entity.MessageVO;
import com.sbp.message.entity.TopicRegistry;
import com.sbp.message.api.service.MessageService;
import com.sbp.message.send_message.MessageSender;
import com.sbp.message.service.cs.MessageCSImpl;
import com.sbp.message.service.cs.TopicRegistryCSImpl;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Service;

import java.util.*;

@Service("messageService")
public class MessageServiceImpl implements MessageService {
    private static final Logger logger = LoggerFactory.getLogger(MessageServiceImpl.class);

    @Autowired
    private TopicRegistryCSImpl topicRegistryCSImpl;
    @Autowired
    private MessageCSImpl messageCSImpl;

    @Autowired
    private MessageSender messageSender;
    @Autowired
    private ErrorNotifyServiceImpl errorNotifyServiceImpl;


    private Message createMessageWithStatus(Message message, MessageStatus messageStatus) {
        message.setId(String.valueOf(IdGenerator.generateId()));
        message.setStatus(messageStatus.getValue());

        // 初始化的时候就保存需要消费的消费端
        //kafka.admin.AdminClient;
        TopicRegistry topicRegistry = topicRegistryCSImpl.selectTopicListByDestination(message.getDestination());
        if (null != topicRegistry) {
            message.setConsumers(topicRegistry.getConsumers());
            message.setConsumersDescribe(topicRegistry.getConsumersDescribe());
        } else {
            message.setConsumers(0);
            message.setConsumersDescribe("");
            logger.warn("message has no consumer, message={}", JSON.toJSONString(message));
        }

        return messageCSImpl.insertMessage(message);
    }
    @Override
    public Message createMessage(Message message) {
        return createMessageWithStatus(message, MessageStatus.unconfirmed);
    }
    // 创建并发送消息
    @Override
    public Message createAndSendMessage(Message message) {
        message = createMessageWithStatus(message, MessageStatus.sending);
        messageSender.sendMessage(message);
        return message;
    }


    // 修改消息状态和错误日志
    @Override
    public int updateMessageStatus(String messageId, MessageStatus messageStatus, String errorMessage, String errorLog) {
        return messageCSImpl.updateMessageStatus(messageId, messageStatus, errorMessage, errorLog);
    }


    @Override
    public int tryEndMessageWithConsumeSucceed(Message message, String appName) {
        // 合法消费者
        if (CommonUtils.resolveIsValidConsumer(message.getConsumersDescribe(), appName)) {
            // 如果未消费, acknowledges++, acknowledges_describe 追加 appName
            int updateRowCount = messageCSImpl.updateMessageWithConsumeSucceedById(message.getId(), appName);
            if (updateRowCount <= 0)
                return 0;

            // 如果 acknowledges >= consumers, 则将消息正常结束
            int deleteRowCount = messageCSImpl.deleteMessageByIdIfAllAcknowledged(message.getId());
            if (deleteRowCount > 0)
                return deleteRowCount;

            // 尝试异常结束
            return messageCSImpl.endMessageByIdWithBusinessFailed(message.getId(), MessageStatus.dead);
        }

        // 意外消费者
        else {
            return messageCSImpl.addSucceedUnexpectedConsumer(message.getId(), appName);
        }
    }
    @Override
    public int tryEndMessageWithBusinessFailed(Message message, String appName, String errorMessage, String errorLog) {
        try {
            // 合法消费者
            if (CommonUtils.resolveIsValidConsumer(message.getConsumersDescribe(), appName)) {
                try {
                    messageCSImpl.updateErrorLogById(message.getId(), errorMessage, errorLog);
                } catch (Exception e) {
                    logger.warn("messageId="+message.getId(), e);
                }

                int updateRowCount = messageCSImpl.updateMessageWithBusinessFailedById(message.getId(), appName);
                if (updateRowCount <= 0)
                    return 0;

                // 尝试异常结束
                return messageCSImpl.endMessageByIdWithBusinessFailed(message.getId(), MessageStatus.dead);
            }

            // 意外消费者
            else {
                return messageCSImpl.addFailedUnexpectedConsumer(message.getId(), appName, errorLog);
            }
        } finally {
            errorNotifyServiceImpl.notify("业务异常", errorMessage, message);
        }
    }

    @Override
    public int updateTopicRegistry(String topic, String appName) {
        return topicRegistryCSImpl.updateTopicRegistry(topic, appName);
    }


    @Override
    public int updateMessageContent(String msgId, String msgContent) {
        return messageCSImpl.updateContentById(msgId, msgContent);
    }


    @Override
    public Message queryMessage(String msgId) {
        return messageCSImpl.selectById(msgId);
    }

    @Override
    public boolean killMessage(String msgId) {
        Message msg = queryMessage(msgId);
        if (null == msg)
            return false;

        return 1 == messageCSImpl.updateMessageStatus(msgId, MessageStatus.dead, null, null);
    }

    @Override
    public boolean closeMessage(String msgId) {
        return 1 == messageCSImpl.updateMessageStatus(msgId, MessageStatus.end, null, null);
    }


    private static void removeBusinessFailedConsumer(Message msg) {
        if (msg.getBusinessFailed() > 0) {
            ArrayList<String> list = CommonUtils.splitByComma(msg.getAcknowledgesDescribe());
            StringBuffer sb = new StringBuffer(",");
            for (String acknowledgeDescribeInDB : list) {
                if (!CommonUtils.isBusinessFailedConsumer(acknowledgeDescribeInDB)) {
                    sb.append(acknowledgeDescribeInDB);
                    sb.append(',');
                }
            }

            msg.setAcknowledgesDescribe(sb.toString());
            msg.setBusinessFailed(0);
        }
    }
    private Message reliveMessageToStatus(String msgId, MessageStatus messageStatus) {
        Message msg = messageCSImpl.selectByIdAndStatus(msgId, MessageStatus.dead);
        if (null == msg)
            return null;

        msg.status = messageStatus.getValue();
        msg.retry = CommonUtils.MAX_RETIES;   // 重试次数设为最大 标记为复活
        removeBusinessFailedConsumer(msg);
        if (1 != messageCSImpl.updateById(msg))
            return null;
        return msg;
    }
    @Override
    public boolean reliveMessage(String msgId) {
        return null != reliveMessageToStatus(msgId, MessageStatus.confirmed);
    }
    @Override
    public boolean reliveAndSendMessage(String msgId) {
        Message msg = reliveMessageToStatus(msgId, MessageStatus.sending);
        if (null == msg)
            return false;
        messageSender.sendMessage(msg);
        return true;
    }


    @Override
    public Message confirmAndSendMessage(Message message) {
        message.setStatus(MessageStatus.sending.getValue());
        if (messageCSImpl.updateUnconfirmedMessage(message) > 0) {
            messageSender.sendMessage(message);
            return message;
        } else {
            logger.warn("messageId={}, confirm message failed", message.getId());
            return null;
        }
    }
    @Override
    public Message confirmAndSendMessage(String msgId) {
        Message msg = messageCSImpl.selectByIdAndStatus(msgId, MessageStatus.unconfirmed);
        if (null == msg)
            return null;

        return confirmAndSendMessage(msg);
    }

    @Override
    public int confirmAndSendMessages(List<Message> messageList) {
        if (null == messageList || messageList.isEmpty())
            return 0;

        messageList = messageCSImpl.confirmMessages(messageList, MessageStatus.sending);

        for (Message message : messageList) {
            messageSender.sendMessage(message);
        }

        return messageList.size();
    }
    @Override
    public int confirmAndSendMessagesById(List<String> messageIdList) {
        List<Message> messageList = messageCSImpl.selectByIdsAndStatus(messageIdList, MessageStatus.unconfirmed);
        return confirmAndSendMessages(messageList);
    }


    @Override
    public Message sendMessage(String msgId) {
        Message message = queryMessage(msgId);
        if (null == message)
            return null;

        messageSender.sendMessage(message);
        return message;
    }
    @Override
    public Message sendMessage(Message msg) {
        messageSender.sendMessage(msg);
        return msg;
    }
    @Override
    public int sendMessages(List<Message> msgList) {
        int res = 0;

        for (Message msg : msgList) {
            messageSender.sendMessage(msg);
            ++res;
        }

        return res;
    }
    @Override
    public int sendMessagesById(List<String> msgIdList) {
        Map<String, Object> parameter = new HashMap<>();
        parameter.put("idList", msgIdList);

        List<Message> msgList = messageCSImpl.selectByIdsAndStatus(msgIdList, null);
        return sendMessages(msgList);
    }


    @Override
    public int removeMessages(List<String> msgIdList) {
        return messageCSImpl.removeMessages(msgIdList);
    }


    @Override
    public Page<Message> queryMessagePage(Map<String, Object> parameter, Integer pageIndex, Integer pageSize) {
        return messageCSImpl.queryMessageList(parameter, pageIndex, pageSize);
    }
    @Override
    public Page<MessageVO> queryMessageVOPage(Map<String, Object> parameter, Integer pageIndex, Integer pageSize) {
        Page<Message> messagePage =  messageCSImpl.queryMessageList(parameter, pageIndex, pageSize);
        List<MessageVO> data = new ArrayList<>();
        for (Message dbData : messagePage.getElements()) {
            data.add(MessageVO.fromMessage(dbData));
        }
        return new Page<>(messagePage.getTotalCount(), messagePage.getPageSize(), messagePage.getPageIndex(), data);
    }

    @Override
    public Page<Message> findUnconsumedMessage(Integer pageIndex, Integer pageSize) {
        return messageCSImpl.page(MessageStatus.unconfirmed, pageIndex, pageSize);
    }

    @Override
    public List<String> findUnconsumedMessageId(Integer pageIndex, Integer pageSize) {
        return messageCSImpl.idPage(MessageStatus.unconfirmed, pageIndex, pageSize);
    }

}
